package com.nepu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.nepu.gmall.realtime.util.DimUtil;
import com.nepu.gmall.realtime.util.DruidDSUtil;
import com.nepu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @author chenshuaijun
 * @create 2023-02-18 11:12
 */
public class MyPhoenixSink extends RichSinkFunction<JSONObject> {
    private DruidDataSource druidDataSource = null;
    // 我们需要在open方法中连接Phoenix
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        druidDataSource = DruidDSUtil.createDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        // TODO 1、拿到对应的连接
        DruidPooledConnection connection = druidDataSource.getConnection();
        // TODO 2、取出对应的表名 : sinkTable
        String sinkTable = value.getString("sinkTable");
        // TODO 3、取出对应的数据
        JSONObject data = value.getJSONObject("data");

        // TODO 如果发现当前的操作是修改操作，我们需要删除redis中的数据
        String type = value.getString("type");
        if ("update".equals(type)){
            String key = data.getString("id");
            DimUtil.deleteRedisData(sinkTable.toUpperCase(), key);
        }
        // TODO 4、拼接SQL，将数据写入到Phoenix对应的表中
        try {
            PhoenixUtil.upsertData(connection,sinkTable,data);
        } catch (Exception e) {
            /*
                这里关于维度数据写入的异常是有两种解决办法的：
                    （1）第一种是，直接跳过这个异常，之后如果在别的层中用到了相应的数据，但是无法查到的时候，
                                 我们直接可以mysql中查
                    （2）第二种解决办法是，我们我们在其它层，如果在Phoenix中查不到对应的数据，不去mysql中查，
                        所以这里如果我们的数据没有写入，那么直接就抛出运行时异常，让程序停止。√
             */
            throw new RuntimeException("维度数据写入失败: " + sinkTable);
        } finally {
            // TODO 5、关闭连接,底层其实是还回对应的连接
            try {
                connection.close();
            } catch (Exception e) {
                System.out.println("连接池的对象归还失败.....");
                e.printStackTrace();
            }

        }

    }
}
